package org.ly.service;

import org.ly.dao.CmdbHostRepository;
import org.ly.dao.HostMetricRepository;
import org.ly.dao.TempMessageRepository;
import org.ly.util.LogUtil;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;

@Service
public class KafkaConsumerService {

    private final HostMetricRepository hostMetricRepository;
    private final TempMessageRepository tempMessageRepository;
    private final CmdbHostRepository cmdbHostRepository;
    private final KafkaTemplate<String, String> kafkaTemplate;

    public KafkaConsumerService(HostMetricRepository hostMetricRepository, KafkaTemplate<String, String> kafkaTemplate, TempMessageRepository tempMessageRepository, CmdbHostRepository cmdbHostRepository) {
        this.hostMetricRepository = hostMetricRepository;
        this.kafkaTemplate = kafkaTemplate;
        this.tempMessageRepository = tempMessageRepository;
        this.cmdbHostRepository = cmdbHostRepository;
    }

    @KafkaListener(topics = "${custom-config.beixiang.eops-metric-topic}", groupId = "${spring.kafka.consumer.group-id}")
    @Transactional
    public void eopsMetricTopicListen(String message) throws IOException {
        LogUtil.logDebug("RawEopsMetricMessage：" + message);
        MetricChecker metricChecker = new MetricChecker(hostMetricRepository, tempMessageRepository, cmdbHostRepository);
        metricChecker.nodataCheck(message);
    }

    @KafkaListener(topics = "${custom-config.beixiang.eops-config-topic}", groupId = "${spring.kafka.consumer.group-id}")
    @Transactional
    public void eopsConfigTopicListen(String message) throws IOException {
        LogUtil.logDebug("RawEopsConfigMessage：" + message);
        MetricChecker metricChecker = new MetricChecker(hostMetricRepository, tempMessageRepository, cmdbHostRepository);
        metricChecker.nothingCheck(message);
    }

}
